package com.rem.flink.flink4Sink;


import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.Set;

/**
 * sink redis
 *
 * @author Rem
 * @date 2022/10/9
 */
public class SinkToRedisTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> stream = env.addSource(new ClickSource());

        FlinkJedisSentinelConfig conf = new FlinkJedisSentinelConfig.Builder()
                .setSentinels(Set.of("localhost:6380", "localhost:6381", "localhost:6382"))
                .setMasterName("mymaster")
                .setPassword("123456")
                .build();
        stream.addSink(new RedisSink<>(conf, new MyRedisMapper()));

        env.execute();
    }

    public static class MyRedisMapper implements RedisMapper<Event> {
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "clicks2");
        }

        @Override
        public String getKeyFromData(Event data) {
            return data.getUser();
        }

        @Override
        public String getValueFromData(Event data) {
            return data.getUrl();
        }
    }

}
